package com.fs.iot.service;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.fs.common.base.service.CacheService;
import com.fs.common.utils.OgnlUtils;
import com.fs.common.config.NoticeProcessor;
import com.fs.iot.bean.dto.DeviceDTO;
import com.fs.iot.bean.dto.DeviceDataDTO;
import com.fs.iot.bean.dto.DeviceModelFieldDTO;
import com.fs.iot.bean.dto.DeviceModelPublishedDTO;
import com.fs.iot.data.AlarmCheckService;
import com.fs.utils.GroovyUtils;
import com.fs.utils.JsUtils;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 原始工况数据处理服务
 * 非Flink的模式
 *
 * @author  LiuQi 2025/3/24-19:20
 * @version V1.0
 **/
@Service
@ConditionalOnProperty(prefix = "iot.flink", name = "enabled", havingValue = "false")
@Slf4j
public class KafkaOriginDataProcessor implements NoticeProcessor {
    @Autowired
    private CacheService cacheService;

    @Autowired
    private DeviceModelPublishedEntityService deviceModelPublishedEntityService;

    @Autowired
    private DeviceEntityService deviceService;

    @Autowired
    private DeviceDataEntityService deviceDataEntityService;

    @Autowired
    private DeviceHistoryDataService historyDataService;

    @Autowired
    private AlarmCheckService alarmCheckService;

    @Value("${iot.kafka.topic.processed}")
    private String processedTopic;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // 设备信息缓存
    private static final Map<String, DeviceDTO> deviceMap = new Hashtable<>();

    // 模型字段信息缓存
    private static final Map<String, Map<String, DeviceModelFieldDTO>> modelFieldMap = new Hashtable<>();

    @PostConstruct
    public void loadAllCache() {
        // 系统启动时，加载所有设备及字段缓存
        List<DeviceDTO> devices = deviceService.queryBuilder().eq("status", 1).query();
        if (!CollectionUtils.isEmpty(devices)) {
            devices.forEach(device -> deviceMap.put(device.getCode(), device));
        }

        List<DeviceModelPublishedDTO> publishedModels = deviceModelPublishedEntityService.findAll();
        publishedModels.forEach(model -> {
            String modelId = model.getId();
            List<DeviceModelFieldDTO> fields = model.getFields();
            Map<String, DeviceModelFieldDTO> map = fields.stream().collect(Collectors.toMap(DeviceModelFieldDTO::getCode, f -> f));
            modelFieldMap.put(modelId, map);
        });
    }

    @Override
    public void process(ConsumerRecord<String, String> record) {
        String type = record.key();
        if ("device".equals(type)) {
            // 设备通知
            DeviceDTO device = JSONObject.parseObject(record.value(), DeviceDTO.class);
            deviceMap.put(device.getCode(), device);
        } else if ("model".equals(type)) {
            DeviceModelPublishedDTO model = JSONObject.parseObject(record.value(), DeviceModelPublishedDTO.class);
            String modelId = model.getId();
            List<DeviceModelFieldDTO> fields = model.getFields();
            Map<String, DeviceModelFieldDTO> map = fields.stream().collect(Collectors.toMap(DeviceModelFieldDTO::getCode, f -> f));
            modelFieldMap.put(modelId, map);
        }
    }

    /**
     * 处理消息内容
     */
    @KafkaListener(topics = "${iot.kafka.topic.origin}", groupId = "IOT_PLATFORM", id = "IOT_PLATFORM")
    public void process(String msg) {
        if (StringUtils.isBlank(msg)) {
            return;
        }

        JSONObject obj = JSON.parseObject(msg);
        String deviceCode = obj.getString("__deviceCode");

        // 查询设备字段信息
        DeviceDTO device = deviceMap.get(deviceCode);
        if (null == device) {
            log.error("设备不存在，设备编号：{}", deviceCode);
            return;
        } else if (device.getStatus() != 1) {
            log.error("设备状态不是已发布状态，设备：{}", deviceCode);
            return;
        }

        // 更新设备在线状态
        device.setOnline(true);
        device.setLastUpTime(LocalDateTime.now());
        deviceService.update(device);

        // 处理属性
        Map<String, DeviceModelFieldDTO> fieldMap = modelFieldMap.get(device.getModelId());
        if (null == fieldMap) {
            log.error("根据设备查询设备字段为空，设备编号：{}", deviceCode);
            return;
        }

        Map<String, Object> workDataMap = new HashMap<>(16);
        List<DeviceDataDTO> result = this.processMsg(device, fieldMap, obj, workDataMap);

        // 保存结果
        saveResult(result);

        // 推送加工后数据队列
        kafkaTemplate.send(processedTopic, device.getTenantId(), JSON.toJSONString(workDataMap));

        // 报警规则校验
        alarmCheckService.check(workDataMap);
    }

    /**
     * 保存数据
     * @param list 数据列表
     */
    private void saveResult(List<DeviceDataDTO> list) {
        // 使用replace into 进行插入或更新
        deviceDataEntityService.batchUpsert(list);

        // 保存历史工况
        historyDataService.batchInsert(list);
    }

    /**
     * 数据转换处理
     */
    private List<DeviceDataDTO> processMsg(DeviceDTO device, Map<String, DeviceModelFieldDTO> fieldMap,
                                           final Map<String, Object> obj,
                                           Map<String, Object> workDataMap) {
        List<DeviceDataDTO> dataList = new ArrayList<>(16);
        String cacheKey = "iot.data." + device.getCode();
        JSONObject lastValue = cacheService.get(cacheKey, JSONObject.class);
        cacheService.cache(cacheKey, obj);

        long ts = MapUtils.getLongValue(obj, "__ts");

        workDataMap.put("__modelId", device.getModelId());
        workDataMap.put("__deviceId", device.getId());
        workDataMap.put("__deviceCode", device.getCode());
        workDataMap.put("__gatewayId", device.getGatewayId());
        workDataMap.put("__ts", ts);

        LocalDateTime timestamp = LocalDateTime.ofEpochSecond(ts / 1000, (int) (ts % 1000), ZoneOffset.ofHours(8));

        fieldMap.keySet().forEach(key -> {
            Object value = obj.get(key);
            DeviceModelFieldDTO field = fieldMap.get(key);

            // 看字段是否需要进行计算
            if (1 == field.getType()) {
                // 计算字段
                Map<String, Object> metadata = field.getMetadata();
                String scriptType = MapUtils.getString(metadata, "scriptType", "js");
                String script = MapUtils.getString(metadata, "script");
                if (StringUtils.isBlank(script)) {
                    return;
                }

                Object result = null;
                if ("js".equals(scriptType) || "javascript".equals(scriptType)) {
                    result = JsUtils.execute(script, obj, lastValue);
                } else if ("groovy".equals(scriptType)){
                    result = GroovyUtils.execute(script, obj, lastValue);
                } else if ("ognl".equals(scriptType)) {
                    result = OgnlUtils.execute(script, obj);
                }

                if (null == result) {
                    return;
                }

                value = result;
            } else if (null == value) {
                return;
            }

            DeviceDataDTO dataDTO = new DeviceDataDTO();
            dataList.add(dataDTO);
            dataDTO.setParamValue(value.toString());
            dataDTO.setDeviceCode(device.getCode());
            dataDTO.setCreateTime(LocalDateTime.now());
            dataDTO.setTs(timestamp);
            dataDTO.setParamCode(key);
            dataDTO.setDeviceId(device.getId());

            // 以设备编号+字段编码做主键存储
            dataDTO.setId(device.getCode() + "-" + field.getCode());

            workDataMap.put(key, value);
        });

        return dataList;
    }
}
